package com.lagou.cl.client;

import com.alibaba.fastjson.JSON;
import com.lagou.cl.client.handler.AccountClientHandler;
import com.lagou.cl.config.Constant;
import com.lagou.cl.config.RpcProperties;
import com.lagou.cl.config.ZooKeeperProperties;
import com.lagou.cl.encode.RpcEncoder;
import com.lagou.cl.model.RpcRequest;
import com.lagou.cl.serializer.JSONSerializer;
import com.lagou.cl.util.Obj2ByteUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.lang.reflect.Proxy;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;


/**
 * @author 程  林
 * @date 2020-05-23 18:55
 * @description //
 * @since V1.0.0
 */
@Component
public class RpcConsumer {

	private Logger logger = LoggerFactory.getLogger(RpcConsumer.class);

	@Autowired
	private RpcProperties rpcProperties;

	//private static ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

	private AccountClientHandler accountClientHandler;

	private Bootstrap bootstrap;
	private EventLoopGroup eventExecutors = new NioEventLoopGroup();

	/**
	 * 每个服务端（不同的服务节点）都需要建立一个连接
	 */
	private static Map<String, Channel> channelMap = new ConcurrentHashMap<>();


	/**
	 * 提供服务的父节点
	 */
	private String producerZnodeName = "/" + Constant.PRODUCER_TYPE;

	/**
	 * 负载均衡的父节点
	 */
	private String loadBalancePath = "/" + Constant.LOADBALANCE;

	@Autowired
	private CuratorFramework curatorFramework;

	@Autowired
	private ZooKeeperProperties zkProperties;

	public void startUp() throws Exception {
		//连接zk服务端
		this.curatorFramework.start();
		this.curatorFramework.blockUntilConnected();
		logger.info("服务器提供端：{} ，已连接上zookeeper服务端：{}", rpcProperties.getProducerName(), zkProperties.getZkHost() + ":" + zkProperties.getPort());

		//初始化Netty客户端
		initClient();

		//获取所有的子节点，每个子节点就是一个Netty服务端，需要建立一个连接
		List<String> children = getChildren();
		//创建socket连接
		//2.客户端启动时，从Zookeeper中获取所有服务提供端节点信息，客户端与每一个服务端都建立连接
		createChannel(children);

		//监听服务
		//3.服务端重新上线，客户端能感知到，并且与重新上线的服务端重新建立连接
		//4.某个服务端下线后，Zookeeper注册列表会自动剔除下线的服务端节点，客户端与下线的服务端断开连接
		listenersOfChild();

	}

	/**
	 * 客户端启动时，从Zookeeper中获取所有服务提供端节点信息，客户端与每一个服务端都建立连接
	 * <p>
	 * 与每个服务端建立连接
	 *
	 * @param providers 服务端在zk服务器上的节点路径
	 */
	private void createChannel(List<String> providers) {

		providers.forEach(producerNode -> {
			if (!channelMap.containsKey(producerNode)) {

				String producerNodeData = null;
				try {
					Stat stat = new Stat();
					byte[] znodeData = this.curatorFramework.getData().storingStatIn(stat).forPath(producerNode);
					if (znodeData != null && znodeData.length > 0) {
						producerNodeData = new String(znodeData);
					}
				} catch (Exception e) {
					e.printStackTrace();
				}

				if (Objects.nonNull(producerNodeData)) {
					try {
						String[] ipPort = producerNodeData.split(":");
						logger.info("~~~~~~~~~~~~~~~~ 创建连接：{}", producerNode);
						//连接
						//bootstrap.connect(rpcProperties.getHost(), rpcProperties.getPort()).sync();
						channelMap.put(producerNode, this.bootstrap.connect(ipPort[0], Integer.parseInt(ipPort[1])).sync().channel());
					} catch (InterruptedException e) {
						logger.error(e.getMessage(), e);
					}
				}
			}
		});
	}

	/**
	 * 服务端重新上线，客户端能感知到，并且与重新上线的服务端重新建立连接
	 * 某个服务端下线后，Zookeeper注册列表会自动剔除下线的服务端节点，客户端与下线的服务端断开连接
	 * <p>
	 * 监听子节点
	 *
	 * @throws Exception
	 */
	private void listenersOfChild() throws Exception {

		PathChildrenCache childrenCache = new PathChildrenCache(this.curatorFramework, this.producerZnodeName, true);
		PathChildrenCacheListener childrenCacheListener = (client, event) -> {
			logger.info("事件：{}", JSON.toJSONString(event));
			ChildData data = event.getData();
			switch (event.getType()) {
				case CHILD_ADDED://添加子节点
					String znodeData = new String(data.getData());
					String[] ipPort = znodeData.split(":");
					logger.info("zk服务端新增服务 -- 当前子节点路径：{} ，当前子节点的数据：{}", data.getPath(), znodeData);
					if (!channelMap.containsKey(data.getPath())) {
						logger.info("~~~~~~~~~~~~~~~~ 创建连接：{}", data.getPath());
						channelMap.put(data.getPath(), this.bootstrap.connect(ipPort[0], Integer.parseInt(ipPort[1])).sync().channel());
					}
					break;
				case CHILD_REMOVED://移除子节点
					String znodeDatb = new String(data.getData());
					String[] ipPortb = znodeDatb.split(":");
					logger.info("zk服务端移除服务 -- 当前子节点路径：{} ，当前子节点的数据：{}", data.getPath(), znodeDatb);
					if (channelMap.containsKey(data.getPath())) {
						logger.info("~~~~~~~~~~~~~~~~ 移除连接：{}", data.getPath());
						Channel channel = channelMap.get(data.getPath());
						channel.close();
						channelMap.remove(data.getPath());
					}
					break;
				case CHILD_UPDATED://子节点的数据变化
					String znodeDataa = new String(data.getData());
					String[] ipPorta = znodeDataa.split(":");
					logger.info("zk服务端 服务数据变化 -- 当前子节点路径：{} ，当前子节点的数据：{}", data.getPath(), znodeDataa);
					if (channelMap.containsKey(data.getPath())) {
						logger.info("~~~~~~~~~~~~~~~~ {} 连接的数据发生变化", data.getPath());
						Channel channel = channelMap.get(data.getPath());
						channel.close();
						channelMap.put(data.getPath(), this.bootstrap.connect(ipPorta[0], Integer.parseInt(ipPorta[1])).sync().channel());
					}
					break;
				default:
					break;
			}
		};
		childrenCache.getListenable().addListener(childrenCacheListener);
		childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
	}

	private void initClient() {
		if (this.accountClientHandler == null) {
			this.accountClientHandler = new AccountClientHandler();
		}

		if (this.bootstrap == null) {
			this.bootstrap = new Bootstrap();
		}

		this.bootstrap.group(eventExecutors)
				.channel(NioSocketChannel.class)
				.option(ChannelOption.TCP_NODELAY, true)
				.handler(new ChannelInitializer<SocketChannel>() {
					@Override
					protected void initChannel(SocketChannel ch) throws Exception {
						ChannelPipeline pipeline = ch.pipeline();

						//对RpcRequest编码
						pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
						//对RpcResponse解码
						//pipeline.addLast(new RpcDecoder(RpcResponse.class, new JSONSerializer()));
						pipeline.addLast(new StringDecoder());

						pipeline.addLast(accountClientHandler);
					}
				});

	}


	/**
	 * @param serviceClass 被调用的服务类
	 * @return obj
	 */
	public Object createProxy(final Class<?> serviceClass) {
		return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class[]{serviceClass}, (proxy, method, args) -> {

			RpcRequest request = new RpcRequest();
			String uuid = UUID.randomUUID().toString();
			request.setRequestId(uuid);
			request.setClassName(method.getDeclaringClass().getName());
			request.setMethodName(method.getName());
			request.setParameterTypes(method.getParameterTypes());
			request.setParameters(args);

			logger.info("requestId:{} ,请求服务器", uuid);
			sentMsg(request);
			Thread.sleep(1000);
			return getData(uuid);

		});

	}

	private void sentMsg(RpcRequest request) throws Exception {

		if (!channelMap.isEmpty()) {
			List<Channel> channels = new ArrayList<>(channelMap.values());

			//当客户端发起调用，每次都选择最后一次响应时间短的服务端进行服务调用，如果时间一致，随机选取一个服务端进行调用，从而实现负载均衡

			//  /NETTY-PRODUCER-1
			List<String> loadBalancePaths = this.curatorFramework.getChildren().forPath(this.loadBalancePath);
			//没有负载均衡节点
			if (loadBalancePaths.isEmpty()) {
				logger.info("{} 没有负载均衡的节点,随机选择一个", this.loadBalancePath);
				int size = channels.size();
				int i = new Random().nextInt(size);
				Channel channel = channels.get(i);
				channel.writeAndFlush(request).sync();
			} else {
				//说明最近只有一个被调用
				if (loadBalancePaths.size() == 1) {
					logger.info("{} 负载均衡的节点只有一个：{}", this.loadBalancePath, loadBalancePaths.get(0));
					Channel channel = channelMap.get(this.producerZnodeName + "/" + loadBalancePaths.get(0));
					channel.writeAndFlush(request).sync();
				} else {
					Map<Long, String> longMap = new TreeMap<>();
					//最近都被调用过,选择最后一次响应时间短的服务端进行服务调用，如果时间一致，随机选取一个服务端进行调用，从而实现负载均衡
					//  /ZK-RPC/LOAD-BALANCE/NETTY-PRODUCER-1
					for (String balancePath : loadBalancePaths) {
						String loadPath = this.loadBalancePath + "/" + balancePath;
						Stat stat = new Stat();
						byte[] loadPathTime = this.curatorFramework.getData().storingStatIn(stat).forPath(loadPath);
						if (loadPathTime != null && loadPathTime.length > 0) {
							long lastTime = Obj2ByteUtil.byteArrayToLong(loadPathTime);

							//时间一样，就选用最后一个服务
							longMap.put(lastTime, balancePath);
						}
					}

					logger.info("{} 负载均衡的节点有多个：{}", this.loadBalancePath, longMap);
					if (!longMap.isEmpty()) {
						List<String> collect = new ArrayList<>(longMap.values());
						String loadPath = collect.get(0);
						Channel channel = channelMap.get(this.producerZnodeName + "/" + loadPath);
						channel.writeAndFlush(request).sync();
					}
				}
			}
		}
	}

	private List<String> getChildren() throws Exception {
		List<String> path = this.curatorFramework.getChildren().forPath(this.producerZnodeName);

		return path.stream().map(this::joinPaths).collect(Collectors.toList());

	}

	/**
	 * 添加上的父节点
	 *
	 * @param childPath
	 * @return
	 */
	private String joinPaths(String childPath) {
		return this.producerZnodeName + "/" + childPath;
	}

	private static Map<String, Object> dataMap = new LinkedHashMap<String, Object>();

	private Object getData(String requestId) {
		logger.info("requestId:{} ,响应数据：{}", requestId, dataMap.get(requestId));
		return dataMap.get(requestId);
	}

	public static Map<String, Object> getDataMap() {
		return dataMap;
	}


	/**
	 * 求Map<K,V>中Value(值)的最小值
	 *
	 * @param map
	 * @return
	 */
	public static Object getMinValue(Map<String, Long> map) {
		if (map == null)
			return null;
		Collection<Long> c = map.values();
		Object[] obj = c.toArray();
		Arrays.sort(obj);
		return obj[0];
	}

	/**
	 * 求Map<K,V>中Value(值)的最大值
	 *
	 * @param map
	 * @return
	 */
	public static Object getMaxValue(Map<String, Long> map) {
		if (map == null)
			return null;
		int length = map.size();
		Collection<Long> c = map.values();
		Object[] obj = c.toArray();
		Arrays.sort(obj);
		return obj[length - 1];
	}

}
